package org.fujene.index;

import java.io.IOException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.logging.Logger;

import org.fujene.communicate.local.LocalPeerSystem;
import org.fujene.structure.Constants;
import org.fujene.structure.Enumerates;
import org.fujene.structure.IndexQueue;
import org.msgpack.MessagePackObject;
import org.msgpack.object.NilType;
import org.msgpack.object.RawType;
import org.msgpack.rpc.Request;
import org.msgpack.rpc.Server;
import org.msgpack.rpc.loop.EventLoop;

/**
 * 書き込みのためのシステムです。
 * レプリケーション非対応バージョン。
 */
public class IndexingSystem{
	// Concurrent化改造

	public boolean stopFlag = false;
	private LocalPeerSystem myPeer = null;
	private ExecutorService exec = null;
	private Logger logger = null;

	public IndexingSystem(LocalPeerSystem myPeer, Logger logger){
		// 待ちうけスレッド起動
		exec = Executors.newSingleThreadExecutor();
		exec.execute(new IndexMainService());
		// index用
		this.myPeer = myPeer;
		this.logger = logger;
	}

	// RPC対応
	public class IndexRPCHandler{
		public void index(Request req) throws Exception{
			MessagePackObject obj = req.getArguments();
			MessagePackObject[] args = obj.asArray();
			if (args.length < 1 || !args[0].isRawType()){
				req.sendError(RawType.create("Error(index): No index word"));
				logger.severe("Error(index): No index word");
			}

			String indexQuery = args[0].asString();

			myPeer.setNewIndex(new IndexQueue(Enumerates.INDEX, indexQuery));

			req.sendResult(NilType.create());
		}

		public void read(Request req){
			myPeer.readMemory();
			req.sendResult(NilType.create());
		}

		public void save(Request req){
			myPeer.saveMemory();
			req.sendResult(NilType.create());
		}

		public void stop(Request req){
			myPeer.setStopper(true);
			this.notifyAll();
			exec.shutdownNow();
			req.sendResult(NilType.create());
		}
	}

	public class IndexMainService implements Runnable{
		public IndexMainService(){}

		@Override
		public void run(){
			logger.fine("IndexMainService started.");
			// RPC対応
			Server server = new Server(EventLoop.defaultEventLoop());
			try{
				server.serve(new IndexRPCHandler());
				server.listen(Constants.INDEX_PORT);

				logger.fine("Index server port " + Constants.INDEX_PORT + " has established.");

				synchronized(this){
					this.wait();
				}

				logger.fine("IndexMainService ended.");

				server.close();
			}
			catch(InterruptedException ignored){}
			catch(IOException e){
				e.printStackTrace();
			}
		}
	}
}
